package com.kcht.itsed.mws.subscribescheduler.comm;

import com.kcht.itsed.mws.extendmodel.jpa.BufferedReportEntity;
import com.kcht.itsed.mws.persist.dao.IBufferedReportDao;
import com.kcht.itsed.mws.persist.dao.ISubscriberDao;
import com.kcht.itsed.mws.subscribescheduler.biz.IConcreteReportSender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Optional;

@Component
@Slf4j
public class CachedReportSenderDecorator implements IConcreteReportSender,Runnable {

	private IConcreteReportSender concreteSender;
	private int cacheCount; //缓存数量，超过了，就顶掉最旧的。

	@Autowired
	private IBufferedReportDao bufferedReportDao;
	@Autowired
	private ISubscriberDao subscriberDao;

	@Autowired
	@Qualifier("aleThreadPool")
	private ThreadPoolTaskExecutor mes;
	private String url;
	
	private boolean sending = false;
	
	//初始化缓存发送调度
	private void startCacheSchedule() {
		mes.execute(this);
	}

	
	public void initContainer(String url, IConcreteReportSender concreteSender, int cacheCount) {
		this.url = url;
		this.concreteSender = concreteSender;
		this.cacheCount = cacheCount;
	}
	
	@Override
	public void send(InputStream reportIS) throws ReportSendFailException {
		BufferedReportEntity be = new BufferedReportEntity();
		be.setSubscriberUrl(url);

		byte[] buff = new byte[1024]; //buff用于存放循环读取的临时数据 
		int rc = 0; 
		try(ByteArrayOutputStream swapStream = new ByteArrayOutputStream()) {
			while ((rc = reportIS.read(buff, 0, 1024)) > 0)swapStream.write(buff, 0, rc); 
			be.setReportBytes(swapStream.toByteArray());

			if(bufferedReportDao.getBufferedCount(url) >= cacheCount) { //缓存数量已满，直接搞死最老的
				bufferedReportDao.fetchLastBufferedReportEntity(url).ifPresent((entity)->{
					bufferedReportDao.deleteById(entity.getId());
				});
			}
			bufferedReportDao.insert(be);
			synchronized(concreteSender) {
				if(sending)concreteSender.notifyAll();
				else startCacheSchedule();
			}
		} catch (IOException e) {
			log.error("",e);
		}
	}

	@Override
	public ReportSendingState getSendingState() {
		return concreteSender.getSendingState();
	}

	@Override
	public void run() {
		sending = true;
		while(sending) {
			try {
				if(!subscriberDao.isSubscriberAlive(url))sending = false;
				Optional<BufferedReportEntity> bufferedReportEntity = bufferedReportDao.fetchLastBufferedReportEntity(url);
				if(bufferedReportEntity.isPresent()) {
					BufferedReportEntity entity = bufferedReportEntity.get();
					ByteArrayInputStream bais = new ByteArrayInputStream(entity.getReportBytes());
					concreteSender.send(bais);
					bufferedReportDao.deleteById(entity.getId());
				}else {
					synchronized(concreteSender) {
						concreteSender.wait(60000);
					}
				}
				
			}catch(Exception e) {
				log.error("",e);
			}
		}
	}

}
